package com.xzx.flink.streamapi.transform;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/17 21:45
 */
public class Transform_08_Rescale {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Integer> source = env.addSource(new RichParallelSourceFunction<Integer>() {
            private boolean running = true;

            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                Random random = new Random();
                while (running) {
                    int value = random.nextInt(4);
                    //将奇数发送到索引为 1 的并行子任务
                    // 将偶数发送到索引为 0 的并行子任务
                    if (value % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                        ctx.collect(value);
                    }
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });
        source.print("上游");
        source.rescale().print("下游").setParallelism(4);

        env.execute(Transform_08_Rescale.class.getSimpleName());
    }
}
